#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sqlite3.h>
#include <unistd.h>
#include <signal.h>
#include "MQTTClient.h"
#include "cJSON.h"
#include "shmem.h"
#include "msg_queue_peer.h"
#include "data.h"
#include <time.h>
#define ADDRESS "tcp://localhost:1883"
#define CLIENTID "ExampleClientSub"
#define TOPIC_RECV "/app/stm32/down"
#define TOPIC_SEND "/app/stm32/up"
#define QOS 1
#define TIMEOUT 10000L
#define FILEROAD "/mnt/config/table.json"
volatile MQTTClient_deliveryToken deliveredtoken;
struct shm_param shm;
FILE *fp = NULL;
struct device *p = NULL; // 共享内存首地址
int timeNum = 0;         // 周期上报刷新时间间隔
int upType = 0;          // 上报模式 0-不上报，客户端主动采集；1-变化上报，即连续2次值不相等；2-周期上报
int upFlag = 0;          // 主动采集模式客户端是否请求标志位
sqlite3 *db;             // 数据库句柄
int Modec = 0;           // 客户端修改模式标志位
char bufrecv[1024] = "";
void delivered(void *context, MQTTClient_deliveryToken dt);
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message);
void connlost(void *context, char *cause);
void control_dev(char *msg);
int deserialization();
char *serialization();
void timeoutHandler();
void historyfind(char *msg,char **recvss);
void changeMode(char *msg);
char *oldData = NULL;
int sqlite_init();
char *errmsg = NULL;
MQTTClient client;
MQTTClient_deliveryToken token;
MQTTClient_message pubmsg = MQTTClient_message_initializer;
int main(int argc, char *argv[])
{
    
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;
    int ch;
    shm_init(&shm, "haha", 1024);
    p = shm_getaddr(&shm);
    printf("共享内存首地址:%p\n",p);
    if (p == NULL)
    {
        printf("共享内存打开失败\n");
        return -1;
    }
    MQTTClient_create(&client, ADDRESS, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    MQTTClient_subscribe(client, TOPIC_RECV, QOS);
    sqlite_init();
    if (deserialization() == 2)
    {
        printf("等待上位机下发点表\n");
        sleep(3);
        char *args[] = {"./test",NULL};
        execv(args[0], args);
    }
    char *oldData = serialization();
    char *buf = serialization();
    pubmsg.payload = buf;                 // 消息正文
    pubmsg.payloadlen = (int)strlen(buf); // 消息正文长度
    pubmsg.qos = QOS;                     // 通信质量
    pubmsg.retained = 0;
    MQTTClient_publishMessage(client, TOPIC_SEND, &pubmsg, &token); // 发布消息 指定客户端 主题
    rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);      // 超时检测
    printf("Message with delivery token %d delivered\n", token);
    // 每分钟上传数据到历史记录
    signal(SIGALRM, timeoutHandler);
    alarm(3);
    while (1)
    {
        if (upType == 0)
        {
            if (upFlag)
            {
                buf = serialization();
                pubmsg.payload = buf;                                           // 消息正文
                pubmsg.payloadlen = (int)strlen(buf);                           // 消息正文长度
                MQTTClient_publishMessage(client, TOPIC_SEND, &pubmsg, &token); // 发布消息 指定客户端 主题
                rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);      // 超时检测
                printf("Message with delivery token %d delivered\n", token);
                upFlag = 0;
            }
        }
        else if (upType == 1)
        {
            sleep(timeNum);
            buf = serialization();
            pubmsg.payload = buf;                                           // 消息正文
            pubmsg.payloadlen = (int)strlen(buf);                           // 消息正文长度
            MQTTClient_publishMessage(client, TOPIC_SEND, &pubmsg, &token); // 发布消息 指定客户端 主题
            rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);      // 超时检测
            printf("Message with delivery token %d delivered\n", token);
        }
        else if (upType == 2)
        {
            sleep(1);
            buf = serialization();
            if (strcmp(oldData, buf))
            {
                pubmsg.payload = buf;                                           // 消息正文
                pubmsg.payloadlen = (int)strlen(buf);                           // 消息正文长度
                MQTTClient_publishMessage(client, TOPIC_SEND, &pubmsg, &token); // 发布消息 指定客户端 主题
                rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);      // 超时检测
                printf("Message with delivery token %d delivered\n", token);
                oldData = buf;
            }
        }
    }
    MQTTClient_unsubscribe(client, TOPIC_RECV);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    return rc;
}

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int i;
    char *payloadptr;
    char *historybuf;
    printf("Message arrived\n");
    printf("     topic: %s\n", topicName);
    printf("   message: ");
    payloadptr = message->payload;
    for (i = 0; i < message->payloadlen; i++)
    {
        bufrecv[i] = *payloadptr;
        putchar(*payloadptr++);
    }
    bufrecv[i] = '\0';
    // 接收到字符串反序列化
    cJSON *root;
    root = cJSON_Parse(bufrecv);
    cJSON *type = cJSON_GetObjectItem(root, "type");
    if (type == NULL)
    {
        printf("eqweqeqeq\n");
        return -1;
    }
    int typechose = type->valueint;
    cJSON_Delete(root);
    printf("dwqewqeqweqeqwe\n");
    switch (typechose)
    {
    case 1: // 手动采集模式采集数据请求
        upFlag = 1;
        break;
    case 2: // 设备控制命令
        control_dev(bufrecv);
        break;
    case 3: // 模式修改命令
        changeMode(bufrecv);
        break;
    case 4: // 历史数据请求命令
        historyfind(bufrecv,&historybuf);
        pubmsg.payload = historybuf;                                           // 消息正文
        pubmsg.payloadlen = (int)strlen(historybuf);                           // 消息正文长度
        MQTTClient_publishMessage(client, TOPIC_SEND, &pubmsg, &token); // 发布消息 指定客户端 主题
        MQTTClient_waitForCompletion(client, token, TIMEOUT);      // 超时检测
        printf("Message with delivery token %d delivered\n", token);
        break;
    default:
        break;
    }
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}
// 反序列化，读取点表信息
int deserialization()
{
    struct device *temp = p;
    struct stat st;
    stat(FILEROAD, &st);
    // 读取点表
    char *buf = (char *)malloc(st.st_size);
    if (buf == NULL)
    {
        printf("开辟空间失败\n");
        return -1;
    }
    FILE *fp = fopen(FILEROAD, "r+");
    if (fp == NULL)
    {
        perror("err\n");
        return -1;
    }
    printf("文件大小:%ld\n", st.st_size);
    if (!st.st_size)
    {
        printf("null\n");
        return 2;
    }
    fread(buf, 1, st.st_size, fp);
    // 反序列化
    cJSON *root;
    root = cJSON_Parse(buf);
    cJSON *report = cJSON_GetObjectItem(root, "report");
    cJSON *type = cJSON_GetObjectItem(report, "type");
    upType = type->valueint;
    cJSON *period = cJSON_GetObjectItem(report, "period");
    timeNum = period->valueint;
    printf("上报模式: %d  刷新时间: %d \n", upType, timeNum);
    // 获取设备节点
    cJSON *stm32 = cJSON_GetObjectItem(root, "stm32");
    cJSON *stm32Data = cJSON_GetObjectItem(stm32, "data");
    cJSON *modbus = cJSON_GetObjectItem(root, "modbus");
    cJSON *modbusData = cJSON_GetObjectItem(modbus, "data");
    // 设备数量存放
    temp->stm32Num = cJSON_GetArraySize(stm32Data);
    printf("stm32设备数量:%d\n", temp->stm32Num);
    temp->modbusNum = cJSON_GetArraySize(modbusData);
    printf("modbus设备数量:%d\n", temp->modbusNum);
    // 设备数据存放
    struct data *dataptr = (struct data *)(temp + 1);
    for (int i = 0; i < temp->stm32Num; i++)
    {
        cJSON *item = cJSON_GetArrayItem(stm32Data, i);
        cJSON *key = cJSON_GetObjectItem(item, "key");
        dataptr->key = key->valueint;
        cJSON *name = cJSON_GetObjectItem(item, "name");
        strcpy(dataptr->name, name->valuestring);
        cJSON *type = cJSON_GetObjectItem(item, "type");
        dataptr->type = type->valueint;
        dataptr++;
    }
    for (int i = 0; i < temp->modbusNum; i++)
    {
        cJSON *item = cJSON_GetArrayItem(modbusData, i);
        cJSON *key = cJSON_GetObjectItem(item, "key");
        dataptr->key = key->valueint;
        cJSON *name = cJSON_GetObjectItem(item, "name");
        strcpy(dataptr->name, name->valuestring);
        cJSON *addr = cJSON_GetObjectItem(item, "addr");
        dataptr->addr = addr->valueint;
        cJSON *type = cJSON_GetObjectItem(item, "type");
        dataptr->type = type->valueint;
        dataptr++;
    }
    fclose(fp);
    cJSON_Delete(root);
    free(buf);
    buf = NULL;
    return 0;
}

// 序列化，将共享内存的内容取出来序列化
char *serialization()
{
    cJSON *root = cJSON_CreateObject();
    cJSON *stm32 = cJSON_CreateObject();
    cJSON *stm32Data = cJSON_CreateArray();
    cJSON *modbus = cJSON_CreateObject();
    cJSON *modbusData = cJSON_CreateArray(); 
    cJSON *type = cJSON_CreateObject();
    cJSON_AddItemToObject(root,"type",cJSON_CreateNumber(1));
    struct data *datap = (struct data *)(p + 1);
    for (int i = 0; i < p->stm32Num; i++)
    {
        cJSON *item = cJSON_CreateObject();
        cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(datap->key));
        cJSON_AddItemToObject(item, "name", cJSON_CreateString(datap->name));
        cJSON_AddItemToObject(item, "type", cJSON_CreateNumber(datap->type));
        switch (datap->type)
        {
        case 1:
            cJSON_AddItemToObject(item, "value", cJSON_CreateNumber(datap->val.b_val));
            break;
        case 2:
            cJSON_AddItemToObject(item, "value", cJSON_CreateNumber(datap->val.i_val));
            break;
        case 3:
            cJSON_AddItemToObject(item, "value", cJSON_CreateNumber(datap->val.f_val));
            break;
        default:
            break;
        }
        datap++;
        cJSON_AddItemToArray(stm32Data, item);
    }
    cJSON_AddItemToObject(stm32, "data", stm32Data);
    cJSON_AddItemToObject(root, "stm32", stm32);
    for (int i = 0; i < p->modbusNum; i++)
    {
        cJSON *data = cJSON_CreateArray();
        cJSON *item = cJSON_CreateObject();
        cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(datap->key));
        cJSON_AddItemToObject(item, "name", cJSON_CreateString(datap->name));
        cJSON_AddItemToObject(item, "type", cJSON_CreateNumber(datap->type));
        cJSON_AddItemToObject(item, "addr", cJSON_CreateNumber(datap->addr));
        switch (datap->type)
        {
        case 1:
            cJSON_AddItemToObject(item, "value", cJSON_CreateNumber(datap->val.b_val));
            break;
        case 2:
            cJSON_AddItemToObject(item, "value", cJSON_CreateNumber(datap->val.i_val));
            break;
        case 3:
            cJSON_AddItemToObject(item, "value", cJSON_CreateNumber(datap->val.f_val));
            break;
        default:
            break;
        }
        cJSON_AddItemToArray(modbusData, item);
        datap++;
    }
    cJSON_AddItemToObject(modbus, "data", modbusData);
    cJSON_AddItemToObject(root, "modbus", modbus);
    char *json_str = cJSON_Print(root);
    cJSON_Delete(root);
    return json_str;
}

// 设备控制命令处理函数
void control_dev(char *msg)
{
    struct message rec;
    rec.datatype = 100;
    cJSON *root;
    root = cJSON_Parse(msg);
    cJSON *device = cJSON_GetObjectItem(root, "device");
    if (device == NULL)
    {
        printf("eewqewqeqweqeqeqweqeqe\n");
        return ;
    }
    
    cJSON *data = cJSON_GetObjectItem(root, "data");
    if (device == NULL)
    {
        printf("eewqewqeqweqeqeqweqeqe\n");
        return ;
    }
    cJSON *key = cJSON_GetObjectItem(data, "key");
    if (device == NULL)
    {
        printf("eewqewqeqweqeqeqweqeqe\n");
        return ;
    }
    cJSON *type = cJSON_GetObjectItem(data, "type");
    if (device == NULL)
    {
        printf("eewqewqeqweqeqeqweqeqe\n");
        return ;
    }
    cJSON *val = cJSON_GetObjectItem(data, "val");
    if (device == NULL)
    {
        printf("eewqewqeqweqeqeqweqeqe\n");
        return ;
    }
    rec.key = key->valueint;
    rec.type = type->valueint;
    switch (rec.type)
    {
    case 1:
        rec.val.b_val = atoi(val->valuestring);
        break;
    case 2:
        rec.val.i_val = atoi(val->valuestring);
        break;
    case 3:
        rec.val.f_val = strtof(val->valuestring, NULL);
        break;
    default:
        break;
    }
    if (!strcmp(device->valuestring, "stm32"))
    {
        msg_queue_send("stm32", &rec, sizeof(rec), 0);
    }
    else
    {
        msg_queue_send("modbus", &rec, sizeof(rec), 0);
    }
    cJSON_Delete(root);
}

// 上报模式修改命令
void changeMode(char *msg)
{
    cJSON *root;
    root = cJSON_Parse(msg);
    cJSON *data = cJSON_GetObjectItem(root, "data");
    cJSON *type = cJSON_GetObjectItem(data, "type");
    if (type->valueint == 1)
    {
        cJSON *period = cJSON_GetObjectItem(data, "period");
        timeNum = period->valueint;
    }
    upType = type->valueint;
    cJSON_Delete(root);
    struct stat st;
    stat(FILEROAD, &st);
    // 读取点表
    char recv[2048] = "";
    FILE *fp = fopen(FILEROAD, "r");
    if (fp == NULL)
    {
        perror("err\n");
        return;
    }
    fread(recv, 1, st.st_size, fp);
    fclose(fp);
    root = cJSON_Parse(recv);
    cJSON *report = cJSON_GetObjectItem(root, "report");
    cJSON_ReplaceItemInObject(report, "type", cJSON_CreateNumber(upType));
    cJSON_ReplaceItemInObject(report, "period", cJSON_CreateNumber(timeNum));
    strcpy(recv, cJSON_Print(root));
    fp = fopen(FILEROAD, "w+");
    if (fp == NULL)
    {
        perror("err\n");
        return;
    }
    fwrite(recv, 1, strlen(recv), fp);
    cJSON_Delete(root);
    fclose(fp);
    char *args[] = {"./test",NULL};
    execv(args[0], args);
}
// 数据库初始化
int sqlite_init()
{
    if (sqlite3_open("./history.db", &db) != SQLITE_OK)
    {
        // stderr:标准错误
        fprintf(stderr, "open err:%s\n", sqlite3_errmsg(db));
        return -1;
    }
    printf("数据库打开成功\n");
    // 数据库操作
    char *errmsg = NULL;
    // 测量人体各项指标历史记录表   时间戳 人体温度 心率 血氧含量
    if (sqlite3_exec(db, "create table historypeo(time long,peo_tem float,huart float,oxygen float);", NULL, NULL, &errmsg) != 0)
    {
        printf("create historypeo err:%s\n", errmsg);
    }
    printf("人体指标历史记录表创建成功\n");
    return 0;
}
// 定时器结束处理函数
void timeoutHandler()
{
    struct sqldata data_peo;
    char sql[256] = "";
    data_peo.time = time(NULL);
    struct data *his = (struct data *)(p + 1);
    for (int i = 0; i < (p->stm32Num + p->modbusNum); i++)
    {
        if (!strcmp(his->name, "BodyTemperature"))
        {
            data_peo.val1 = his->val.f_val;
        }
        else if (!strcmp(his->name, "HeartRate"))
        {
            data_peo.val2 = his->val.f_val;
        }
        else if (!strcmp(his->name, "oxygen_data"))
        {
            data_peo.val3 = his->val.f_val;
        }
        his++;
    }
    printf("人体参数: %.2f %.2f %.2f\n", data_peo.val1, data_peo.val2, data_peo.val3);
    sprintf(sql, "insert into historypeo values(%ld,%.2f,%.2f,%.2f)", data_peo.time, data_peo.val1, data_peo.val2, data_peo.val3);
    if (sqlite3_exec(db, sql, NULL, NULL, &errmsg) != 0)
    {
        printf("人体指标数据插入失败:%s\n", errmsg);
    }
    alarm(60);
}
// 历史记录查询
void historyfind(char *msg,char **recvss)
{
    
    char sql[512] = "";
    long start = 0;
    long end = 0;
    int hang = 0;
    int lie = 0;
    char **result = NULL;
    cJSON *root = cJSON_Parse(msg);
    cJSON *data = cJSON_GetObjectItem(root,"data");
    cJSON *item1 = cJSON_GetArrayItem(data,0);
    cJSON *item2 = cJSON_GetArrayItem(data,1);
    printf("历史记录查询\n");
    start = item1->valueint;
    end = item2->valueint;
    cJSON_Delete(root);
    printf("起始：%ld\n",start);
    printf("结尾：%ld\n",end);
    root = cJSON_CreateObject();
    cJSON_AddItemToObject(root,"type",cJSON_CreateNumber(2));
    cJSON *datanum = cJSON_CreateObject();
    int num = 4;
    int number = 0;
    sprintf(sql,"select * from historypeo where time BETWEEN %ld AND %ld",start,end);
    if (sqlite3_get_table(db, sql, &result, &hang, &lie, &errmsg) != 0)
    {
            printf("get table err:%s\n", errmsg);
    }
    data = cJSON_CreateArray();
    for (int i = 0; i < hang; i++)
    {
        cJSON *item = cJSON_CreateObject();
        cJSON_AddItemToObject(item, "time", cJSON_CreateString(result[num++]));
        cJSON_AddItemToObject(item, "tem_per", cJSON_CreateString(result[num++]));
        cJSON_AddItemToObject(item, "huart", cJSON_CreateString(result[num++]));
        cJSON_AddItemToObject(item, "oxygen", cJSON_CreateString(result[num++]));
        cJSON_AddItemToArray(data, item);
        number++;
    }
    cJSON_AddItemToObject(root,"data",data);
    cJSON_AddItemToObject(root,"datanum",cJSON_CreateNumber(number));
     *recvss = cJSON_Print(root);
    cJSON_Delete(root);
}

